feat: support micro batching alternative at source by vaibhavtiwari33 · Pull Request #3450 · numaproj/numaflow

vaibhavtiwari33 · 2026-06-02T15:27:12Z

Currently source reads 1 batch of messages at a time and waits for ack/nack of the whole batch of messages it read and sent downstream.
This is fine for pipeline which can read multiple batches (read ahead)
In case of monovertex, it essentially blocks reading of additional messages until the current batch gets fully acked/nacked (which can be delayed in cases of large tail latency).

What this PR does / why we need it

This PR aims to add an opt-in feature for monovertex to enable per message ack loop (instead of per batch acking)

Added a field in monovertex spec to enable streaming (off by default)
When streaming is enabled, acks for messages are tracked individually instead of in a batch
Max in-flight messages are still capped using concurrency
Semaphores (==concurrency) are distributed per message instead of per batch
Most of the new logic lives in streaming_source in source.rs where we're doing the same operations as before, but on a per message level.

Implemented spec:

kind: MonoVertex
metadata:
  name: my-mvtx
spec:
  streaming: true
...

Related issues

#3452

Testing

Unit tests and an e2e test was added
Tested with running a normal ud source with and without streaming enabled

Tracing Test:

Streaming disabled; transformer returns results

Streaming enabled: transformer returns results

Streaming disabled; transformer omits results

Streaming enabled; transformer omits results

Special notes for reviewers

This is a monovertex only feature currently since pipeline already supports reading multiple batches.

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

…. Implement watermark handling Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

codecov · 2026-06-02T15:44:56Z

Codecov Report

❌ Patch coverage is 87.37728% with 90 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.67%. Comparing base (38f3306) to head (639f079).

Files with missing lines	Patch %	Lines
rust/numaflow-core/src/source.rs	86.77%	82 Missing ⚠️
rust/numaflow-core/src/shared/create_components.rs	33.33%	6 Missing ⚠️
pkg/apis/numaflow/v1alpha1/mono_vertex_types.go	0.00%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3450      +/-   ##
==========================================
+ Coverage   82.55%   82.67%   +0.12%     
==========================================
  Files         307      307              
  Lines       77618    78169     +551     
==========================================
+ Hits        64077    64629     +552     
+ Misses      12984    12983       -1     
  Partials      557      557

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

vigith · 2026-06-02T22:31:34Z

+                    None => vec![msg_handle],
+                    Some(transformer) => {
+                        match transformer
+                            .transform_batch(vec![msg_handle], cln_token.clone(), None)


did you test tracing with this?

I haven't tested end to end tracing with this change but all similar tracing changes have been made as the normal path, so should be good. I'll test this branch as well.

One more change that needs to be made in a separate PR is message level metrics (replacing batch level metrics for source)

You're right, I missed creating spans for transformer in this branch. Fixing this.

Updated, ty for catching this!

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

Signed-off-by: Vigith Maurice <vigith@gmail.com>

vigith

added a WARN if streaming and readAhead are set simultaneously

vaibhavtiwari33 · 2026-06-04T03:26:15Z

+                "streaming=true supersedes read_ahead=true; read_ahead is ignored in streaming mode"
+            );
+        }
+


Since currently streaming is a mvtx only feature, I'll move this inside the following block:

if self.streaming { self.streaming_source( pipeline_labels, mvtx_labels, bypass_router, messages_tx, cln_token, ) .await }

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

…d bypass router are Some Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

vaibhavtiwari33 added 8 commits June 2, 2026 09:50

feat: add Streaming *bool field to MonoVertexSpec

d14227a

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

feat(validator): reject streaming=true with built-in Kafka source

1fd732e

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

feat: streaming MonoVertex source — continuous read and per-message ack

da1d9a2

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

fix: acquire-after-read + inline dispatch for streaming source

73ac357

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

feat: gate source-side batch metrics off under streaming

650089c

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

docs: streaming MonoVertex user-guide page with Kafka data-loss warning

a37a897

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

test(e2e): streaming MonoVertex end-to-end test

932cf72

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

feat: Update streaming documentation. Extract streaming source method…

8372613

…. Implement watermark handling Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

vaibhavtiwari33 changed the title ~~feat: remove micro batching at source~~ feat: support micro batching alternative at source Jun 2, 2026

vaibhavtiwari33 added 2 commits June 2, 2026 14:29

chore: update comments

140041e

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

chore: make codegen

2d09eab

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

vaibhavtiwari33 marked this pull request as ready for review June 2, 2026 18:38

vaibhavtiwari33 requested review from vigith, whynowy and yhl25 as code owners June 2, 2026 18:38

vigith reviewed Jun 2, 2026

View reviewed changes

vaibhavtiwari33 and others added 2 commits June 3, 2026 11:35

fix: initialize and track spans per message for streaming

d826fa9

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

fix: emit end span if all are filtered

02351d8

Signed-off-by: Vigith Maurice <vigith@gmail.com>

vaibhavtiwari33 requested a review from vigith June 4, 2026 01:58

chore: warn if streaming and readAhead is set to true simultaneously

9f2cc19

Signed-off-by: Vigith Maurice <vigith@gmail.com>

vigith approved these changes Jun 4, 2026

View reviewed changes

vaibhavtiwari33 commented Jun 4, 2026

View reviewed changes

chore: move read_ahead warn within the if streaming true block

970e7dc

Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

vaibhavtiwari33 requested a review from syayi June 4, 2026 21:49

chore: tests to cover streaming source scenarios where transformer an…

639f079

…d bypass router are Some Signed-off-by: Vaibhav Tiwari <vaibhav.tiwari33@gmail.com>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: support micro batching alternative at source#3450

feat: support micro batching alternative at source#3450
vaibhavtiwari33 wants to merge 15 commits into
mainfrom
strm-mvtx

vaibhavtiwari33 commented Jun 2, 2026 •

edited

Loading

Uh oh!

codecov Bot commented Jun 2, 2026 •

edited

Loading

Uh oh!

vigith Jun 2, 2026

Uh oh!

vaibhavtiwari33 Jun 3, 2026

Uh oh!

vaibhavtiwari33 Jun 3, 2026

Uh oh!

vaibhavtiwari33 Jun 3, 2026 •

edited

Loading

Uh oh!

vigith left a comment

Uh oh!

vaibhavtiwari33 Jun 4, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

vaibhavtiwari33 commented Jun 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What this PR does / why we need it

Related issues

Testing

Tracing Test:

Special notes for reviewers

Uh oh!

codecov Bot commented Jun 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

vigith Jun 2, 2026

Choose a reason for hiding this comment

Uh oh!

vaibhavtiwari33 Jun 3, 2026

Choose a reason for hiding this comment

Uh oh!

vaibhavtiwari33 Jun 3, 2026

Choose a reason for hiding this comment

Uh oh!

vaibhavtiwari33 Jun 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vigith left a comment

Choose a reason for hiding this comment

Uh oh!

vaibhavtiwari33 Jun 4, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

vaibhavtiwari33 commented Jun 2, 2026 •

edited

Loading

codecov Bot commented Jun 2, 2026 •

edited

Loading

vaibhavtiwari33 Jun 3, 2026 •

edited

Loading